package com.seezoon.infrastructure.tcp.session;

import com.seezoon.infrastructure.error.ErrorCode;
import com.seezoon.infrastructure.exception.Assertion;
import com.seezoon.infrastructure.exception.ExceptionFactory;
import com.seezoon.infrastructure.tcp.codec.JTMessage;
import com.seezoon.infrastructure.tcp.handler.FutureManager;
import com.seezoon.infrastructure.tcp.handler.ResponseFuture;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class Session {

    public final static AttributeKey<Session> key = AttributeKey.valueOf("session");
    // session 超时时间，单位S，默认180s
    private final int timeout = 30 * 60;
    /**
     * 等待回包超时，单位毫秒ms
     */
    private final int defaultReadTimeout = 1000 * 10;
    private final Channel channel;
    /**
     * 终端序列号
     */
    private final String clientNo;
    private final LocalDateTime createTime;
    private final AtomicInteger sequence = new AtomicInteger(0);
    /**
     * 鉴权通过后，放入服务器的 clientId
     */
    private Long clientId;
    private LocalDateTime lastAccessTime;
    private boolean invalidated = false;
    private boolean authed = false;

    public Session(String clientNo, Channel channel) {
        Assertion.notEmpty(clientNo, "clientNo is empty");
        Assertion.notNull(channel, "channel must not be null");
        this.channel = channel;
        channel.attr(key).set(this);
        this.clientNo = clientNo;
        this.createTime = LocalDateTime.now();
        this.lastAccessTime = LocalDateTime.now();
    }

    public void invalidate() {
        this.invalidated = true;
        this.authed = false;
        if (null != channel && channel.isOpen()) {
            channel.close();
        }
    }

    public boolean isInvalidated() {
        boolean result = invalidated || Duration.between(lastAccessTime, LocalDateTime.now()).getSeconds() > timeout;
        if (result) {
            log.info("session invalidated clientNo={}, createTime={}, lastAccessTime={}", clientNo, createTime,
                    lastAccessTime);
        }
        return result;
    }

    public String getClientNo() {
        access();
        return clientNo;
    }

    private void access() {
        this.lastAccessTime = LocalDateTime.now();
    }

    public int nextSequence() {
        access();
        // 无符号short 最大值后，从0 开始
        return sequence.getAndUpdate(current ->
                (current >= 65535) ? 0 : current + 1
        );
    }

    public LocalDateTime getCreateTime() {
        access();
        return createTime;
    }

    public LocalDateTime getLastAccessTime() {
        access();
        return lastAccessTime;
    }

    public boolean isAuthed() {
        access();
        return authed;
    }

    public void setAuthed(Long clientId) {
        Assertion.notNull(clientId, "clientId is null");
        access();
        this.clientId = clientId;
        this.authed = true;
    }

    public Long getClientId() {
        return clientId;
    }

    public <T> T send(JTMessage jtMessage, Class<T> responseClazz) {
        return this.send(jtMessage, responseClazz, defaultReadTimeout);
    }

    public <T> T send(JTMessage jtMessage, Class<T> responseClazz, long timeout) {
        Assertion.notNull(jtMessage, "jtMessage must not be null");
        Assertion.notNull(responseClazz, "responseClazz must not be null");
        if (channel == null) {
            throw ExceptionFactory.bizException(ErrorCode.DEVICE_OFFLINE);
        }
        if (!channel.isActive()) {
            log.error("send failed, channel is closed,jtMessage {}", jtMessage);
            throw ExceptionFactory.bizException(ErrorCode.TCP_CHANNEL_CLOSED);
        }

        CompletableFuture<T> future = new CompletableFuture<>();
        FutureManager.getServerFutureManager().add(channel.id().asLongText(), jtMessage.getMsgSerialNo(),
                new ResponseFuture(responseClazz, future, timeout));
        channel.writeAndFlush(jtMessage);
        try {
            return future.get(timeout, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            log.error("get response failed channel:{},jtmsg:{}", channel, jtMessage, e);
            throw ExceptionFactory.bizException(ErrorCode.TCP_MESSAGE_READ_TIMEOUT);
        }
    }
}
